package org.eking.bigdata.hadoop;


import java.io.BufferedInputStream;  
import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;  
import java.io.FileInputStream;  
import java.io.FileOutputStream;  
import java.io.IOException;  
import java.io.InputStream;  
import java.io.OutputStream;
import java.net.MalformedURLException;
import java.net.URI;  
import java.net.URL;

import org.apache.commons.io.output.ByteArrayOutputStream;
import org.apache.hadoop.conf.Configuration;  
import org.apache.hadoop.fs.FSDataInputStream;  
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;  
import org.apache.hadoop.fs.FsUrlStreamHandlerFactory;  
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi.Factory;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.StringUtils;
import org.apache.log4j.Logger; 
public class HdfsTest {

	static final Logger logger = Logger.getLogger(HdfsTest.class);  
//	static {
//		FsUrlStreamHandlerFactory fac = new FsUrlStreamHandlerFactory();
//		if(fac == null)
//		{
//			System.out.println("xxxx\n");
//			
//		}
//		URL.setURLStreamHandlerFactory(fac);
//	}
	public static void main(String[] args) {
		// TOD
		String user;
		user = "bdtst";
		System.setProperty("HADOOP_USER_NAME", user);
		System.out.println("FileSystemTest\n");
		TestInputStream();
		//FileSystemOutTest();
		//FileSystemTest();
		//TestZipLib();
		//TestWriteable();
		//MkdirTest();
		System.out.println("success\n");
	}
	public static void TestAvro(){
	  
	}	
	public static void TestWriteable(){//test serialize and deserialize
		try{
			ByteArrayOutputStream out = new ByteArrayOutputStream();
			DataOutputStream dataout = new DataOutputStream(out);
			IntWritable writable = new IntWritable();
			writable.set(166);
			writable.write(dataout);
			byte[] b = out.toByteArray();
			System.out.print(StringUtils.byteToHexString(b));
			System.out.println("\nend\n");
			dataout.close();
			
			ByteArrayInputStream in = new ByteArrayInputStream(b);
			DataInputStream datain = new DataInputStream(in);
			writable.readFields(datain);
			System.out.println(writable.get());
			datain.close();
			System.out.println("\nend2\n");
		} catch (IOException e) {
			// TODO �Զ����ɵ� catch ��
			e.printStackTrace();
		}
	}
	public static void TestZipLib(){
		/* input stream zip test code
		String CodecClassName="org.apache.hadoop.io.compress.GzipCodec";
		try {
			Class<?> codecClass = Class.forName(CodecClassName);
			Configuration conf = new Configuration();
			CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, conf);
			CompressionOutputStream out = codec.createOutputStream(System.out);
			InputStream in = new ByteArrayInputStream("hello zx".getBytes());
			IOUtils.copyBytes(System.in, out, conf);
			out.finish();
		} catch (ClassNotFoundException | IOException e) {
			// TODO
			e.printStackTrace();
		}
		*/
		
		/*file zip test */
		String uri = "";
		Configuration conf = new Configuration();
		InputStream in = null;
		OutputStream out = null;
		try {
			FileSystem fs = FileSystem.get(URI.create(uri), conf);
			Path inputpath = new Path(uri);
			
			CompressionCodecFactory factory = new CompressionCodecFactory(conf);
			CompressionCodec codec = factory.getCodec(inputpath);
			if(codec == null)
			{
				System.err.println("no codec found\n");
				System.exit(0);
			}
			
			String outputuri = CompressionCodecFactory.removeSuffix(uri, codec.getDefaultExtension());
					
			in = codec.createInputStream(fs.open(inputpath));
			out = fs.create(new Path(outputuri));
			IOUtils.copyBytes(in, out, conf);
		} catch (IOException e) {
			// TODO 
			e.printStackTrace();
		}finally{
			IOUtils.closeStream(in);
			IOUtils.closeStream(out);
		}
		
	}
	public static void TestInputStream(){
		// TODO
		try{
			
			SecurityManager Sm = new SecurityManager();
			//Sm.checkSetFactory();
		}catch ( SecurityException e) {
			// TODO
			e.printStackTrace();
			return;
		}
		InputStream in = null;
		try{
			FsUrlStreamHandlerFactory fac = new FsUrlStreamHandlerFactory();
			fac.createURLStreamHandler("hdfs");
			URL.setURLStreamHandlerFactory(fac);//ÿ��JVMֻ������һ��
			URL url = new URL("hdfs://10.71.88.13:9001/lele/README.txt");
			in = url.openStream();
//			System.out.print(in.read());
			IOUtils.copyBytes(in, System.out, 4096, false);
			
		} catch (MalformedURLException e) {
			// TODO 
			e.printStackTrace();
		} catch (IOException e) {
			// TODO 
			e.printStackTrace();
		}catch (Error e) {
			// TODO
			e.printStackTrace();
		}finally{
			IOUtils.closeStream(in);
		}
	}
	
	public static void FileSystemTest()
	{
		Configuration conf = new Configuration();
		String Uri = "hdfs://10.71.88.13:9001/lele/test.txt";
		//InputStream in = null;
		FSDataInputStream in = null;
		try {
			FileSystem fs = FileSystem.get(URI.create(Uri), conf);
			in = fs.open(new Path(Uri));
			in.seek(0);
			in.getPos();
			IOUtils.copyBytes(in, System.out, 4096, false);
		} catch (IOException e) {
			// TODO 
			e.printStackTrace();
		}finally{
			IOUtils.closeStream(in);		
		}
	}
	
	public static void FileSystemOutTest()
	{
		long pos = 0;
		String Uri = "hdfs://10.71.88.13:9001/lele/test.txt";
		FSDataOutputStream out = null;
		Configuration conf = new Configuration();
		conf.set("dfs.client.block.write.replace-datanode-on-failure.policy", "NEVER");
		conf.set("dfs.client.block.write.replace-datanode-on-failure.enable", "true");
		InputStream in = null;
		String instr = null;
		FileSystem fs;
		try {
			
			fs = FileSystem.get(URI.create(Uri), conf);
			out = fs.append(new Path(Uri));
			out.writeChars("xxx ");
			out.write("hello".getBytes());
			pos = out.getPos();
			instr = String.format("seek is %d\n", pos);
			//in = new ByteArrayInputStream(instr.getBytes());
			//in.close();
			//IOUtils.copyBytes(in, out, conf);
			
		} catch (IOException e) {
			// TODO �Զ����ɵ� catch ��
			e.printStackTrace();
		}finally{
			IOUtils.closeStream(out);	
			
		}
	}
	
	public static void MkdirTest(){
		Configuration conf = new Configuration();
		FileSystem fs;
		String Uri = "hdfs://10.71.88.13:9001/lele/";
		try {
			
			fs = FileSystem.get(URI.create(Uri), conf);
			/*
			Path path = new Path("hdfs://127.0.0.1:9000/lele/test");
			if(!fs.exists(path)){
				fs.mkdirs(path);
			}
			*/
			
			FSDataOutputStream out = fs.create(new Path("hdfs://10.71.88.13:9001/lele/test/aa.txt"));
			out.writeChars("test");
			out.close();
			
			FileStatus stat = fs.getFileStatus(new Path("hdfs://10.71.88.13:9001/lele/test/aa.txt"));
			
			System.out.println(stat.getPath());
			System.out.println(stat.isDir());
			System.out.println(stat.getLen());
			System.out.println(stat.getOwner());
			//MiniClusters cluster = null;
		} catch (IOException e) {
			// TODO �Զ����ɵ� catch ��
			e.printStackTrace();
		}
	}
}
